Potential issue with string interning.

This is fixed in #146. Originally by syncing on this.queues, later by replacing with ConcurrentHashMap

may be ConcurrentHashMap?

Done in #146

nit: return null over return queue for better readability primarily because there is a null comparision before return. Also, if queues is a ConcurrentHashMap, it might be
EventQueue queue = queues.get(taskId);
return queue != null ? queue.tap() : null;

Done in #146

ConcurrentHashMap maybe since there is a lot of async handling.

Done in #146

can use async composition.

I'm not totally clear what you mean here?

I was thinking of something like this

runningAgents.getOrDefault(task.getId(), CompletableFuture.completedFuture(null)).cancel(true); but this is more of a nit for readability, might not have a large gain.

or even

Optional.ofNullable(runningAgents.get(task.getId()))
.ifPresent(task -> task.cancel(true));

Async via CompetableFuture may be?

Not totally clear again :-)

Same as dequeue, public CompletableFuture enqueueEvent(Event event) {..}

@rajeshvelicheti Just wondering whether or not it makes sense to make enqueueEvent async and return a CompletableFuture. With this approach, it doesn't seem like we can guarantee the order in which events get added to the queue.

For example, if enqueueEvent(eventA) is called followed by enqueueEvent(eventB) and if the enqueueEvent method is async and uses something like this code snippet to handle adding the event, it doesn't seem like we can guarantee that eventA will be added to the queue before eventB:

return CompletableFuture.supplyAsync(() -> {
            try {
                semaphore.acquire();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException("Unable to acquire the semaphore to enqueue the event", e);
            }
            queue.add(event);
            // Call toString() since for errors we don't really want the full stacktrace
            log.debug("Enqueued event {} {}", event instanceof Throwable ? event.toString() : event, this);
            return null;
        });

@rajeshvelicheti Please let us know if we are missing something or if you had something else in mind. Thanks!

Async via CompletableFuture may be?

Not clear to me

Thanks Kabir. I thought, public CompletableFuture dequeueEvent(int waitMilliSeconds) throws EventQueueClosedException {...}

might help in this, given how the method is structured is being used.

Looking at the usage, AutoCloseable might help the consumers.

Done in #146

ConcurrentHashMap maybe?

Done in #146

Can be unbounded to handle blocking better, a bounded semaphore to handle back pressure might be effective

I'm a bit wooly on the details, but I have replaced ArrayBlockingQueue with LinkedBlockingQueue. The latter has a constructor where you don't specify the capacity, so I took a guess at this being a better structure from your input. Then I added a fair semaphore to deal with the capacity.

Those changes are in #146.

If I have misunderstood something, please let me know.

I was thinking of something like this

runningAgents.getOrDefault(task.getId(), CompletableFuture.completedFuture(null)).cancel(true); but this is more of a nit for readability, might not have a large gain.

or even

Optional.ofNullable(runningAgents.get(task.getId()))
.ifPresent(task -> task.cancel(true));

what if cf is null? would the queuemanager need to close?

Also, cf !=null could be avoided for readability

A cf means there is an active task being processed. In which case we remove (i.e. 'close') that task from the queuemanager. The queuemanager itself remains open.

That said, having revisited the code, I don't think the future can be null, so I will try removing the null check.

153-175 have same code, might be useful to convert to a common method. Nit

both if and else do the same thing.

There is a TODO with something to investigate from the Python implementation

nit: taskId.equals might be more helpful as there is a not null comparison on taskId already. Potential NPE

Fixed

nit: Potential NPE.

Fixed

nit: heavy constructor

I'll leave this as-is for now. We need to create an id if not specified

nit: heavy constructor

TextPart - is this always expected? would the cast fail in any case?

The method seems too indicate we're explicitly looking for TextParts, and since we're filtering on Kind.TEXT the cast cannot fail

returning a modifiable collection, is this okay? looking at the references might help to return an un-modifiable collection

I've wrapped it in Collections.unmodifiableList() for now.

The main user seems to be the AgentExecutor, and I'm not totally clear if those should be able to add to this list. That said, there is a RequestContext.attachRelatedTask() that AgentExecutor implementors can call.

do we need the tube closed in this case as well?

Good point. I don't remember the exact details of if/how the Python implementation evolved in this area. But looking at the latest code there, it seems to do the equivalent of what you say.

I've updated this as per your suggestion.

potential infinite loop, do we expect the tube to close all the time so there is no infinite loop?

IMO we should listen until the queue is closed (signalled via an EventQueueClosedException), an error is received, or a 'final' event is received.

I believe that is handled correctly?

java.net.URI over string, could help with the slashes.

Addressed in #182

can be private final